package com.cloud.common.broadcast;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

/**
 * 广播处理
 */
@Configuration
@ConditionalOnClass(RedisTemplate.class)
@ConditionalOnMissingClass("com.cloud.gateway.bean.GatewayInfo")
public class BroadcastHandler {
    public static final String TOPIC = "ChannelTopic";
    private static final Map<String, MessageHandler> handlerMap = new HashMap<>();

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(new RedisMessageListener(), new ChannelTopic(TOPIC));
        return container;
    }

    /**
     * 发布消息
     *
     * @param message
     */
    public void publish(BroadcastMessage message) {
        redisTemplate.convertAndSend(TOPIC, message);
    }

    /**
     * 订阅消息
     *
     * @param key
     * @param handler
     */
    public void subscribe(String key, MessageHandler handler) {
        handlerMap.put(key, handler);
    }

    /**
     * 取消订阅
     *
     * @param key
     */
    public void unsubscribe(String key) {
        handlerMap.remove(key);
    }

    /**
     * 消息处理内部类
     */
    class RedisMessageListener implements MessageListener {
        @Override
        public void onMessage(Message message, byte[] pattern) {
            BroadcastMessage broadcastMessage = (BroadcastMessage) redisTemplate.getValueSerializer().deserialize(message.getBody());
            MessageHandler handler = handlerMap.get(broadcastMessage.getKey());
            if (handler != null) {
                handler.handle(broadcastMessage);
            }
        }
    }
}
